package integration.data.task.api.transformation.impl;

import integration.data.task.api.transformation.Transformer;
import integration.modle.exception.IntegrationValidException;
import io.github.bigbird0101.datatransfer.model.element.*;
import io.github.bigbird0101.datatransfer.record.Record;
import lombok.extern.slf4j.Slf4j;

import java.util.Arrays;
import java.util.Map;

import static integration.data.task.api.constant.Constant.*;

/**
 * 数据过滤
 */
@Slf4j
public class DataFilter extends Transformer {

    public DataFilter() {
        setTransformerName(DATA_FILTER);
    }

    @Override
    public Record evaluate(Record record, Map<String, Object> paras) {
        int columnIndex;
        String code;
        String value;

        try {
            if (paras.size() != 3) {
                throw new IntegrationValidException("过滤参数必须是3个");
            }

            columnIndex = (Integer) paras.get(COLUMN);
            code = (String) paras.get(FILTER_TYPE);
            value = (String) paras.get(VALUE);
            if (value == null) {
                throw new IntegrationValidException("过滤参数value不能是null");
            }
        } catch (Exception e) {
            throw new IntegrationValidException("转换参数异常" + "paras:" + Arrays.asList(paras).toString() + " => " + e.getMessage());
        }

        Column column = record.getColumn(columnIndex);

        try {
            /**
             *  {@link integration.data.enums.ruleChain.FilterTypeEnum}
             *     EQ(0, "等于", "="),
             *     NE(1, "不等于", "!="),
             *     GT(2, "大于", ">"),
             *     LT(3, "小于", "<"),
             *     GE(4, "大于等于", ">="),
             *     LE(5, "小于等于", "<="),
             *     LIKE(6, "模糊", "%");
             */
            if ("LIKE".equalsIgnoreCase(code)) {
                return doLike(record, value, column);
            } else if ("NOT_LIKE".equalsIgnoreCase(code)) {
                return doNotLike(record, value, column);
            } else if ("GT".equalsIgnoreCase(code)) {
                return doGreat(record, value, column, false);
            } else if ("LT".equalsIgnoreCase(code)) {
                return doLess(record, value, column, false);
            } else if ("EQ".equalsIgnoreCase(code) || "==".equalsIgnoreCase(code)) {
                return doEqual(record, value, column);
            } else if ("NE".equalsIgnoreCase(code)) {
                return doNotEqual(record, value, column);
            } else if ("GE".equalsIgnoreCase(code)) {
                return doGreat(record, value, column, true);
            } else if ("LE".equalsIgnoreCase(code)) {
                return doLess(record, value, column, true);
            } else {
                throw new IntegrationValidException("不支持的过滤类型:" + code);
            }
        } catch (Exception e) {
            throw new IntegrationValidException("过滤运行异常", e);
        }
    }


    private Record doGreat(Record record, String value, Column column, boolean hasEqual) {

        //如果字段为空，直接不参与比较。即空也属于无穷小
        if (column.getRawData() == null) {
            return record;
        }
        if (column instanceof DoubleColumn) {
            Double ori = column.asDouble();
            double val = Double.parseDouble(value);

            if (hasEqual) {
                if (ori >= val) {
                    return null;
                } else {
                    return record;
                }
            } else {
                if (ori > val) {
                    return null;
                } else {
                    return record;
                }
            }
        } else if (column instanceof LongColumn || column instanceof DateColumn) {
            Long ori = column.asLong();
            long val = Long.parseLong(value);

            if (hasEqual) {
                if (ori >= val) {
                    return null;
                } else {
                    return record;
                }
            } else {
                if (ori > val) {
                    return null;
                } else {
                    return record;
                }
            }
        } else if (column instanceof StringColumn || column instanceof BytesColumn || column instanceof BoolColumn) {
            String ori = column.asString();
            if (hasEqual) {
                if (ori.compareTo(value) >= 0) {
                    return null;
                } else {
                    return record;
                }
            } else {
                if (ori.compareTo(value) > 0) {
                    return null;
                } else {
                    return record;
                }
            }
        } else {
            throw new IntegrationValidException(">=,> 不支持这个字段类型:" + column.getClass().getSimpleName());
        }
    }

    private Record doLess(Record record, String value, Column column, boolean hasEqual) {

        //如果字段为空，直接不参与比较。即空也属于无穷大
        if (column.getRawData() == null) {
            return record;
        }

        if (column instanceof DoubleColumn) {
            Double ori = column.asDouble();
            double val = Double.parseDouble(value);

            if (hasEqual) {
                if (ori <= val) {
                    return null;
                } else {
                    return record;
                }
            } else {
                if (ori < val) {
                    return null;
                } else {
                    return record;
                }
            }
        } else if (column instanceof LongColumn || column instanceof DateColumn) {
            Long ori = column.asLong();
            long val = Long.parseLong(value);

            if (hasEqual) {
                if (ori <= val) {
                    return null;
                } else {
                    return record;
                }
            } else {
                if (ori < val) {
                    return null;
                } else {
                    return record;
                }
            }
        } else if (column instanceof StringColumn || column instanceof BytesColumn || column instanceof BoolColumn) {
            String ori = column.asString();
            if (hasEqual) {
                if (ori.compareTo(value) <= 0) {
                    return null;
                } else {
                    return record;
                }
            } else {
                if (ori.compareTo(value) < 0) {
                    return null;
                } else {
                    return record;
                }
            }
        } else {
            throw new IntegrationValidException("<=,< 不支持这个字段类型:" + column.getClass().getSimpleName());
        }

    }

    /**
     * DateColumn将比较long值，StringColumn，ByteColumn以及BooleanColumn比较其String值
     *
     * @param record
     * @param value
     * @param column
     * @return 如果相等，则过滤。
     */

    private Record doEqual(Record record, String value, Column column) {

        //如果字段为空，只比较目标字段为"null"，否则null字段均不过滤
        if (column.getRawData() == null) {
            if ("null".equalsIgnoreCase(value)) {
                return null;
            } else {
                return record;
            }
        }

        if (column instanceof DoubleColumn) {
            Double ori = column.asDouble();
            double val = Double.parseDouble(value);

            if (ori == val) {
                return null;
            } else {
                return record;
            }
        } else if (column instanceof LongColumn || column instanceof DateColumn) {
            Long ori = column.asLong();
            long val = Long.parseLong(value);

            if (ori == val) {
                return null;
            } else {
                return record;
            }
        } else if (column instanceof StringColumn || column instanceof BytesColumn || column instanceof BoolColumn) {
            String ori = column.asString();
            if (ori.compareTo(value) == 0) {
                return null;
            } else {
                return record;
            }
        } else {
            throw new IntegrationValidException("== 不支持这个字段类型:" + column.getClass().getSimpleName());
        }

    }

    /**
     * DateColumn将比较long值，StringColumn，ByteColumn以及BooleanColumn比较其String值
     *
     * @param record
     * @param value
     * @param column
     * @return 如果不相等，则过滤。
     */
    private Record doNotEqual(Record record, String value, Column column) {

        //如果字段为空，只比较目标字段为"null", 否则null字段均过滤。
        if (column.getRawData() == null) {
            if ("null".equalsIgnoreCase(value)) {
                return record;
            } else {
                return null;
            }
        }

        if (column instanceof DoubleColumn) {
            Double ori = column.asDouble();
            double val = Double.parseDouble(value);

            if (ori != val) {
                return null;
            } else {
                return record;
            }
        } else if (column instanceof LongColumn || column instanceof DateColumn) {
            Long ori = column.asLong();
            long val = Long.parseLong(value);

            if (ori != val) {
                return null;
            } else {
                return record;
            }
        } else if (column instanceof StringColumn || column instanceof BytesColumn || column instanceof BoolColumn) {
            String ori = column.asString();
            if (ori.compareTo(value) != 0) {
                return null;
            } else {
                return record;
            }
        } else {
            throw new IntegrationValidException("== 不支持这个字段类型:" + column.getClass().getSimpleName());
        }
    }

    private Record doLike(Record record, String value, Column column) {
        String orivalue = column.asString();
        if (orivalue != null && orivalue.matches(value)) {
            return null;
        } else {
            return record;
        }
    }

    private Record doNotLike(Record record, String value, Column column) {
        String orivalue = column.asString();
        if (orivalue != null && orivalue.matches(value)) {
            return record;
        } else {
            return null;
        }
    }
}
